package org.example.kpi;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class KPIBrowser {

	public static class KPIBrowserMapper extends Mapper<Object, Text, Text, IntWritable> {
		private IntWritable one = new IntWritable(1);
		private Text word = new Text();

		/**
		 * 对输入的键值对进行处理，将值（文本）中的浏览器信息提取出来，并将该信息作为键，将计数器"1"作为值，写入到上下文中。
		 * 该方法是MapReduce框架中的Map阶段的实现，负责数据的初步处理。
		 *
		 * @param key 输入键，该示例中未使用。
		 * @param value 输入值，预期为文本格式，包含需要解析的浏览器信息。
		 * @param context 上下文对象，用于输出处理后的键值对到下一阶段的处理。
		 * @throws IOException 如果发生I/O错误。
		 * @throws InterruptedException 如果执行过程中被中断。
		 */
		@Override
		public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
		    // 从文本中过滤提取浏览器信息
		    KPI kpi = KPI.filterBroswer(value.toString());
		    // 检查提取的浏览器信息是否有效
		    if (kpi.isValid()) {
		        // 设置键为提取的浏览器信息
		        word.set(kpi.getHttp_user_agent());
		        // 将处理结果写入上下文
		        context.write(word, one);
		    }
		}
	}

	public static class KPIBrowserReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
		private IntWritable result = new IntWritable();

		/**
		 * 对于给定的键，将与该键相关联的所有值求和，并将结果写回到上下文中。
		 * @param key 文本类型的键，表示一组值的共同标识符。
		 * @param values 与键相关联的整数可迭代集合，表示需要进行求和操作的一组数值。
		 * @param context 用于读取输入和写入输出的上下文环境。
		 * @throws IOException 如果发生I/O错误。
		 * @throws InterruptedException 如果执行被中断。
		 */
		@Override
		public void reduce(Text key, Iterable<IntWritable> values, Context context)
				throws IOException, InterruptedException {
			int sum = 0; // 初始化求和变量
			for (IntWritable value : values) {
				sum += value.get(); // 遍历并累加所有与键相关联的值
			}
			result.set(sum); // 将求和结果设置到结果对象中
			context.write(key, result); // 将键值对写入到输出上下文中
		}
	}

	/**
	 * 主函数：计算网页访问日志中各浏览器的KPI（关键性能指标）。
	 * @param args 不使用外部传入参数。
	 * @throws Exception 处理Hadoop作业过程中可能抛出的异常。
	 */
	public static void main(String[] args) throws Exception {
	    // 定义输入和输出路径
	    String input = "hdfs://master:9000/weblog/logfile";
	    String output = "hdfs://master:9000/weblog/kpi/browser";

	    // 配置Hadoop环境
	    Configuration conf = new Configuration();
	    // 创建一个新的作业实例并设置作业名称
	    Job job = Job.getInstance(conf, "KPIBrowser");
	    // 设置作业使用的Jar包
	    job.setJarByClass(KPIPV.class);
	    // 设置Mapper和Reducer的输出类型
	    job.setMapOutputKeyClass(Text.class);
	    job.setMapOutputValueClass(IntWritable.class);

	    job.setOutputKeyClass(Text.class);
	    job.setOutputValueClass(IntWritable.class);

	    // 设置Mapper和Reducer类
	    job.setMapperClass(KPIBrowserMapper.class);
	    // 注释掉Combiner类的设置，但保留代码以便未来可能的需求
	    // job.setCombinerClass(KPIBrowserReducer.class);
	    job.setReducerClass(KPIBrowserReducer.class);

	    // 设置输入和输出文件路径
	    FileInputFormat.addInputPath(job, new Path(input));
	    FileOutputFormat.setOutputPath(job, new Path(output));
	    // 等待作业完成，如果作业失败，则返回
	    if (!job.waitForCompletion(true))
	        return;
	}

}
